package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo20Partitions {
  def main(args: Array[String]): Unit = {
    // RDD分区数的调整
    val conf = new SparkConf()
    conf.setMaster("local[4]") // 间接的影响第一个RDD的分区数
    conf.set("spark.default.parallelism", "8") // 会影响shuffle算子的并行度
    conf.setAppName(this.getClass.getSimpleName.replace("$", ""))
    val sc = new SparkContext(conf)

    // 第一个RDD的分区数默认等于切片的数量，也可以手动指定分区数
    // textFile中有一个minPartitions参数，默认等于 任务并行度和2取最小值
    // 所以当任务并行度为1时，则可以只有一个分区，其它情况并行度超过1时，则最少会有2个分区
    val stuRDD: RDD[String] = sc.textFile("scala/data/stu/students.txt")
    println(s"stuRDD的分区数为:${stuRDD.getNumPartitions}")

    // 子RDD的分区数默认等于父RDD的分区数
    val stuKVRDD: RDD[(String, Int)] = stuRDD.map(line => (line.split(",")(4), 1))
    println(s"stuKVRDD的分区数为:${stuKVRDD.getNumPartitions}")

    // shuffle类的算子可以在调用时手动指定分区数，如果没有指定则考虑是否有参数，如果没有再直接从父类继承
    /**
     * 优先级：
     * 1、手动指定
     * 2、通过参数：spark.default.parallelism指定
     * 3、继承父RDD的分区数
     */
    val clazzCntRDD: RDD[(String, Int)] = stuKVRDD.reduceByKey(_ + _)
    println(s"clazzCntRDD的分区数为:${clazzCntRDD.getNumPartitions}")

    val resRDD: RDD[String] = clazzCntRDD.map(kv => s"${kv._1},${kv._2}")
    println(s"resRDD的分区数为:${resRDD.getNumPartitions}")

    resRDD.foreach(println)

    // 手动的改变分区：repartition、coalesce
    val rdd01: RDD[Int] = sc.parallelize(List(1, 2, 3, 4, 5, 6), 3)
    println(s"rdd01的分区数为：${rdd01.getNumPartitions}")

    // repartition：既可以增加分区，又可以减少分区
    // 实际上也调用了coalesce的方法，只是将shuffle参数设为true
    val rdd01_1: RDD[Int] = rdd01.repartition(4)
    println(s"rdd01_1的分区数为：${rdd01_1.getNumPartitions}")
    val rdd01_2: RDD[Int] = rdd01.repartition(2)
    println(s"rdd01_2的分区数为：${rdd01_2.getNumPartitions}")


    // coalesce默认shuffle参数为false，只能用于减少分区
    val rdd01_3: RDD[Int] = rdd01.coalesce(4)
    println(s"rdd01_3的分区数为：${rdd01_3.getNumPartitions}")
    val rdd01_4: RDD[Int] = rdd01.coalesce(2)
    println(s"rdd01_4的分区数为：${rdd01_4.getNumPartitions}")


  }

}
